package cn.edu.ccnu.nlp;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FamilyFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;


public class Corpus {
	
	public static Configuration configuration;
	public static HTablePool pool;
	public static HTable news;
	public static HTable blog;
	public static HTable bbs;
	public static HTable news_stat;
	public static HTable blog_stat;
	public static HTable bbs_stat;
	static {
		configuration = HBaseConfiguration.create();
		
		try{
			pool = new HTablePool(configuration, 1000);
			news = (HTable)pool.getTable("news");
			//blog = (HTable)pool.getTable("blog");
			//bbs = (HTable)pool.getTable("bbs");
			news_stat = (HTable)pool.getTable("news_stat");
			//blog_stat = (HTable)pool.getTable("blog_stat");
			//bbs_stat = (HTable)pool.getTable("bbs_stat");
		}catch(Exception e){
			e.getStackTrace();
		}
	}

	/**
	 * 查询年份的数据，以Map形式返回该年的数据量
	 * @param tableName
	 * @param year
	 * @param yearNum
	 * @return
	 */
	public static Map<String,Integer> QueryByYear(HTable table,String year,Map<String,Integer> yearNum) {
		int sum = 0;
		String year_1 = null;
		ResultScanner rs = null;
		try {
			Filter filter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator(year+".*")); // 查询一年的数据，正则表达式形式
			Scan s = new Scan();
			s.setFilter(filter);
			rs = table.getScanner(s);
			for (Result r : rs) {
				year_1 = (new String(r.getRow())).substring(0,4);                                      //获取年份
				for (KeyValue keyValue : r.raw()) {
					if((new String(keyValue.getFamily())).equals("num"))
						sum += Integer.parseInt(new String(keyValue.getValue()));
				}
			}
			yearNum.put(year_1, sum);
			System.out.println(year_1+":"+sum);
		} catch (Exception e) {
			e.printStackTrace();
		}
		finally{
			rs.close();
		}
		return yearNum;
	}
	
	/**
	 * 查询没一年的数据量
	 * @param tableName
	 * @param years
	 * @return
	 */
	public Map<String,Integer> QueryByYears(HTable table,List<String> years){
		Map<String,Integer> yearNum = new HashMap<String,Integer>();
		for(String year:years){
			yearNum = QueryByYear(table,year,yearNum);
		}
		return yearNum;
		
	}
		
	/**
	 * 单条件按查询，查询多条记录
	 * @param tableName
	 */
	public static Map<String,Integer> QueryByMonth(HTable table,String year,String month,Map<String,Integer> monthNum) {

		int sum = 0;
		String month_1 = null;
		ResultScanner rs = null;
		try {
			Filter filter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator(year+"-"+month+"-.*")); // 查询某年某月的记录数
			Scan s = new Scan();
			s.setFilter(filter);
			rs = table.getScanner(s);
			for (Result r : rs) {
				month_1 = new String(r.getRow()).substring(5,7);                    //获取字符串中的月份                                     //获得月份
				for (KeyValue keyValue : r.raw()) {
					if((new String(keyValue.getFamily())).equals("num"))                                       //当为数量的列时，统计总数
						sum += Integer.parseInt(new String(keyValue.getValue()));
				}
			}
			System.out.println(year+"month:"+month_1+"sum:"+sum);
			monthNum.put(month_1, sum);
		} catch (Exception e) {
			e.printStackTrace();
		}
		finally{
			rs.close();
		}
		return monthNum;
	}
	
	/**
	 * 统计所有月份的数据量
	 * @param tableName
	 * @param year
	 * @param months
	 * @return
	 */
	public Map<String,Integer> QueryByMonths(HTable table,String year,List<String> months){
		Map<String,Integer> monthNum = new HashMap<String,Integer>();
		for(String month:months){
			monthNum = QueryByMonth(table,year,month,monthNum);
		}
		return monthNum;
		
	}
	/**
	 * 统计每一天的数据量
	 * @param tableName
	 */
	public static Map<String,Integer> QueryByDay(HTable table,String year,String month,String day,Map<String,Integer> dayNum) {
			int sum = 0;
			String day_1 = null;
			ResultScanner rs = null;
		try {
			Filter filter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator(year+"-"+month+"-"+day)); // 某年某月某日的数据量统计,注意日期的格式是否正确
			Scan s = new Scan();
			s.setFilter(filter);
			rs = table.getScanner(s);
			for (Result r : rs) {
				day_1 = new String(r.getRow()).substring(8, 10);                   //获取字符串中的日期
				System.out.print(new String(r.getRow())+":"+day_1+":");
				for (KeyValue keyValue : r.raw()) {
//					System.out.println("列：" + new String(keyValue.getFamily())
//							+ "====值:" + new String(keyValue.getValue()));
					if((new String(keyValue.getFamily())).equals("num"))
						sum = Integer.parseInt(new String(keyValue.getValue()));
					System.out.println(sum);
				}
			}
			dayNum.put(day_1, sum);
		} catch (Exception e) {
			e.printStackTrace();
		}
		finally{
			rs.close();
		}
		return dayNum;
	}
	
	/**
	 * 统计某个月所有天数的数据量
	 * @param tableName
	 * @param year
	 * @param month
	 * @param days
	 * @return
	 */
	public Map<String,Integer> QueryByDays(HTable table,String year,String month,List<String> days){
		Map<String,Integer> dayNum = new HashMap<String,Integer>();
		for(String day:days){
			dayNum = QueryByDay(table,year,month,day,dayNum);
		}
		return dayNum;
		
	}
	
	/**
	 * 获取HBase数据库中所有的年份
	 * @param tableName
	 */
	public List<String> getYears(HTable table){
		
		List<String> years = new ArrayList<String>();
		ResultScanner rs = null;
		try {
			Filter filter = new FamilyFilter(CompareOp.GREATER_OR_EQUAL,new BinaryComparator(Bytes.toBytes("2005-01-01"))); //从2005年起始
			Scan scan = new Scan();
			scan.setFilter(filter);
			rs = table.getScanner(scan);
			String year;
			for(Result result:rs){
				year = new String(result.getRow()).substring(0, 4);                                                         //获取年份并进行统计
				if(!years.contains(year)){                                      
					years.add(year);
				}
			}
			
		} catch (IOException e) {
			e.printStackTrace();
		}
		finally{
			rs.close();
		}
		return years;
	}
	
	
	public String getYearString(int current_year, int current_month, int current_day, int current_hour, String corpus) throws Exception
	{
		List<String> years = null;
		if(corpus.equals("news"))
			years = getYears(news_stat);
		if(corpus.equals("blog"))
			years = getYears(blog_stat);
		if(corpus.equals("bbs"))
			years = getYears(bbs_stat);
			
		int start_year = 2005;
		int end_year  = start_year + 11;
		int years_count[] = null;
		double years_score[] = null;
		int total_count = 0;

		String key_count = corpus + "_" + "years_count";
		String key_score = corpus + "_" + "years_score";


		if(years_score==null || years_count==null)
		{
		    years_count = new int[12];
			years_score = new double[12];
			
			Map map = new HashMap();
			
			if(corpus.equals("news"))
				map = QueryByYears(news_stat,years);	
			if(corpus.equals("blog"))
				map = QueryByYears(blog_stat,years);
			if(corpus.equals("bbs"))
				map = QueryByYears(bbs_stat,years);
			
			int max = 0;
			Set set = map.keySet();
			Iterator it = set.iterator();
			while(it.hasNext()){
				String key = (String)it.next();
			try{
				System.out.println(key+":"+map.get(key));
				int y = Integer.parseInt(key);
//				System.out.println("key"+key);
				int c = years_count[y - start_year]  = (Integer)map.get(key);	
				System.out.println("pos:"+(y-start_year)+":value:"+c);
				max = (c > max) ? c : max;
			}catch(Exception e){
			 }
		}

			for(int i = 0; i < years_count.length; i++)
					total_count += years_count[i];

		}



		if( total_count == 0 )
			return "<graph></graph>";
		
			StringBuffer put_time_xml = new StringBuffer();
			put_time_xml.append("<graph  canvasBorderColor='cccccc'  canvasbgColor='cccccc' caption='' subcaption='' xAxisName='年' yAxisMinValue='0' yAxisName='' decimalPrecision='0' formatNumberScale='0' showNames='1' showValues='0'  showAlternateHGridColor='1' AlternateHGridColor='ff5904' divLineColor='ff5904' divLineAlpha='30' alternateHGridAlpha='5' >");


		for(int i = start_year; i <= end_year; i++)
		{
			String link =  String.format("corpus.jsp?c=%s&y=%s&m=%s&d=%s", corpus,  i, 1, 1);
			String color = ( i == current_year ) ? "FF0000" : "ff5904";
			put_time_xml.append( String.format("<set color='%s' name='%s' value='%s'  link='%s'/>", color, i,  years_count[i - start_year] , link) );
		}


		put_time_xml.append("</graph>"); 
		return put_time_xml.toString();

	}

	////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

	public String getMonthString(int current_year, int current_month, int current_day, int current_hour, String corpus) throws Exception
	{
		int months_count[] = null;
		double months_score[] = null;
		int  total_count = 0;

		String key_count = corpus + "_months_count_" + current_year;
		String key_score = corpus + "_months_score_" + current_year;

		if( months_score == null || months_count == null )
		{
			months_count = new int[12];
			months_score = new double[12];
			List<String> months = new ArrayList<String>();
			months.add("01");months.add("02");months.add("03");months.add("04");
			months.add("05");months.add("06");months.add("07");months.add("08");
			months.add("09");months.add("10");months.add("11");months.add("12");
			Map map = new HashMap();
			
			if(corpus.equals("news"))
				map = QueryByMonths(news_stat,""+current_year+"",months);
			if(corpus.equals("blog"))
				map = QueryByMonths(blog_stat,""+current_year+"",months);
			if(corpus.equals("bbs"))
				map = QueryByMonths(bbs_stat,""+current_year+"",months);
			
			
			Set set = map.keySet();
			Iterator it = set.iterator();
			int max = 0;
			while(it.hasNext()){
				String key;
				try{
					key = (String)it.next();
					int m = Integer.parseInt(key );		
					int c = months_count[m - 1]  =  (Integer)map.get(key);	
					max = (c > max) ? c : max;
				}catch(Exception e){
				 }
			}

				for(int i = 0; i < months_count.length; i++)
					total_count += months_count[i];
			
		}

		if( total_count == 0 )
			return "<graph></graph>";

		StringBuffer put_time_xml = new StringBuffer();
			put_time_xml.append("<graph canvasBorderColor='cccccc'  canvasbgColor='cccccc' caption='' subcaption='' xAxisName='月' yAxisMinValue='0' yAxisName='' decimalPrecision='0' formatNumberScale='0' showNames='1' showValues='0'  showAlternateHGridColor='1' AlternateHGridColor='ff5904' divLineColor='ff5904' divLineAlpha='30' alternateHGridAlpha='5' >");

		for(int i = 1; i <= 12; i++)
		{
			String link =  String.format("corpus.jsp?c=%s&y=%s&m=%s&d=%s", corpus, current_year, i, 1);
			String color = ( i == current_month ) ? "FF0000" : "ff5904";
			put_time_xml.append( String.format("<set color = '%s' name='%s' value='%s'  link='%s'/>", color, i,  months_count[i - 1] , link) );
		}
		put_time_xml.append("</graph>"); 

		return put_time_xml.toString();

	}

	////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

	public String getDayString(int current_year, int current_month, int current_day, int current_hour, String corpus)  throws Exception
	{
		
		List<String> days = new ArrayList<String>();
		for(int i = 1;i<=31;i++){                     //保持格式的一致，月和日都要为两位数
			String day = null;
			if(i < 10)
				day = "0"+i;
			else
				day = ""+i;
			days.add(day);
		}
		String current_month_1,current_day_1;            //月份格式的调整
		if(current_month < 10){                           
			current_month_1 = "0"+current_month;
		}
		else current_month_1 = ""+current_month;
//		if(current_day < 10)
//			current_day_1 = "0"+current_day;
//		current_day_1 = ""+current_day;
		
		int days_count[] = null;
		double days_score[] = null;
		int total_count = 0;
		
		String key_count = corpus + "_days_count_" + current_year + "_" + current_month;
		String key_score = corpus + "_days_score_" + current_year + "_" + current_month;
		
		if( days_score == null || days_count == null)
		{
			days_count = new int[31];
			days_score = new double[31];

			Map map = new HashMap();
			if(corpus.equals("news"))
				map = QueryByDays(news_stat,""+current_year,current_month_1,days);
			if(corpus.equals("blog"))
				map = QueryByDays(blog_stat,""+current_year,current_month_1,days);
			if(corpus.equals("bbs"))
				map = QueryByDays(bbs_stat,""+current_year,current_month_1,days);
			
			Set set = map.keySet();
			Iterator it = set.iterator();
			int max = 0;
			while(it.hasNext()){
				try{
					String key = (String)it.next();
					int d = Integer.parseInt(key);		
					int c =days_count[d - 1] =  (Integer)map.get(key);		
					max = (c > max) ? c : max;
					System.out.println(key+":"+c);
				}catch(Exception e){
				 }
			}

				for(int i = 0; i < days_count.length; i++)
					total_count += days_count[i];

				//admin.putInCache(key_count, days_count);
				//admin.putInCache(key_score, days_score);
		}

		if( total_count == 0 )
			return "<graph></graph>";

		StringBuffer put_time_xml = new StringBuffer();
			put_time_xml.append("<graph  canvasBorderColor='cccccc'  canvasbgColor='cccccc' caption='' subcaption='' xAxisName='日' yAxisMinValue='0'  decimalPrecision='0' formatNumberScale='0' showNames='1' showValues='0'  showAlternateHGridColor='1' AlternateHGridColor='ff5904' divLineColor='ff5904' divLineAlpha='30' alternateHGridAlpha='5' >");
		
		for(int i = 1; i <= 31; i++)
		{
			String link =  String.format("corpus.jsp?c=%s&y=%s&m=%s&d=%s", corpus, current_year, current_month, i);
			String color = ( i == current_day ) ? "FF0000" : "ff5904";
			put_time_xml.append( String.format("<set color='%s' name='%s' value='%s'  link='%s'/>",color, i,  days_count[i - 1] , link) );
		}
		put_time_xml.append("</graph>"); 

		return put_time_xml.toString();
	}

	/**
	 * 获取某天的记录数
	 */
	public int currentDayCount(String tableName,int current_year, int current_month, int current_day){
		tableName = tableName+"_stat";
		String month,day;
		int count = 0;
		if(current_month < 10)
			month = "0"+current_month;
		else
			month = ""+current_month;
		
		if(current_day < 10)
			day = "0"+current_day;
		else
			day = ""+current_day;
		try {
			
			HTablePool pool = new HTablePool(configuration, 1000);
			HTable table = (HTable) pool.getTable(tableName);
			Filter filter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator(current_year+"-"+month+"-"+day)); // 某年某月某日的数据量统计,注意日期的格式是否正确
			Scan s = new Scan();
			s.setFilter(filter);
			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				for (KeyValue keyValue : r.raw()) {
						count = Integer.parseInt(new String(keyValue.getValue()));
				}
			}
			System.out.println(count);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return count;
	}
	
	public void getNewsInfo(String tableName,int current_year,int current_month,int current_day,int pageSize){
	
		String month,day;
		if(current_month < 10)
			month = "0"+current_month;
		else
			month = ""+current_month;
		
		if(current_day < 10)
			day = "0"+current_day;
		else
			day = ""+current_day;
		try{
		HTablePool pool = new HTablePool(configuration, 1000);
		HTable table = (HTable) pool.getTable(tableName);
		
		List<Filter> filters = new ArrayList<Filter>();
		
		Filter filter1 = new SingleColumnValueFilter(
				Bytes.toBytes("date"), null, CompareOp.EQUAL, new RegexStringComparator(current_year+"-"+month+"-"+day)); 
		Filter filter2 = new PageFilter(pageSize);
		filters.add(filter1);
		filters.add(filter2);
		FilterList listFilter = new FilterList(filters);
		
		Scan scan = new Scan();
		scan.setFilter(listFilter);
		ResultScanner rs = table.getScanner(scan);
		for(Result r: rs){
				System.out.println(new String(r.getRow()));
				
				KeyValue title = r.getColumnLatest("title".getBytes(), null);
				KeyValue date = r.getColumnLatest("date".getBytes(), null);
				KeyValue source = r.getColumnLatest("source".getBytes(), null);
				System.out.println("url：====值:" + new String(r.getRow()));
				System.out.println("date：====值:" + new String(date.getValue()));
				System.out.println("source：====值:" + new String(source.getValue()));
				System.out.println("title：====值:" + new String(title.getValue()));
//				totalRows++;
//				}
		}
		rs.close();
	}catch(Exception e){
		
	}
	}
	
	/**
	 * 单条件查询,根据rowkey查询唯一一条记录
	 * @param tableName
	 */
	public static void getNews(String tableName,String rowKey) {

		HTablePool pool = new HTablePool(configuration, 1000);
		HTable table = (HTable) pool.getTable(tableName);
		try {
			Get scan = new Get(rowKey.getBytes());// 根据rowkey查询
			Result r = table.get(scan);
			System.out.println("获得到rowkey:" + new String(r.getRow()));
			for (KeyValue keyValue : r.raw()) {
				System.out.println("列：" + new String(keyValue.getFamily())
						+ "====值:" + new String(keyValue.getValue()));
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	public static void main(String[] args) throws Exception
	{
		String[] info = null;
		Corpus corpus = new Corpus();
		List<String> years = new ArrayList<String>();
		years.add("2005");
		years.add("2006");
		years.add("2007");
		years.add("2008");
		years.add("2009");
		years.add("2010");
		years.add("2011");
		years.add("2012");
//		th.createTable("blog", "base-i nfo");
//		th.insertData("news",info);
		
//		corpus.QueryByYears(news_stat,years);
//		th.QueryAll("news_stat");
//		th.QueryByDay("news_stat");
//		th.getYear("news_stat");
//		corpus.QueryByDays("news_stat", "2005","08", days);
//		System.out.println(corpus.getYearString(2005, 1, 1, 22, "news"));
//		System.out.println(corpus.getDayString(2009, 5, 23, 22, "news"));
//		List<String> days = new ArrayList<String>();
//		for(int i = 1;i<=31;i++){
//			String day = null;
//			if(i < 10)
//				day = "0"+i;
//			else
//				day = ""+i;
//			days.add(day);
//		}
//		corpus.QueryByDays("news_stat", "2009","05",days);
		corpus.currentDayCount("news", 2005, 1, 1);
//		corpus.getNewsInfo("news",2005,1,3,10);
//		corpus.getNews("news", "2005-01-01  04德甲最佳阵容：卡恩再受青睐 德利刃不敌巴荷双枪");
	}

}
